package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.client.RedisClient;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import redis.clients.jedis.JedisCluster;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

/**
 * @author Liu.Nanfang
 * @date 2023/3/27 13:24
 * @description 充电主题信号变化监听处理
 */
@Slf4j
public class DwsSignalChangeTask extends BaseTask {
    private static final String DASK_NAME = "dws-signal-change";
    /**
     * 缓存 key
     * 表示：charge:dws:status:vin:
     */
    private static final String DWS_CHARGE_STATUS_KEY = "charge:dws:s:v:";
    /**
     * 缓存时间 1 小时
     */
    private static final long DWS_CHARGE_STATUS_EXPIRE_KEY = 1 * 60 * 60;


    public static void main(String[] args) {
        try {
            ConfigLoader.init(args);
            StreamExecutionEnvironment env = getEnv(DASK_NAME);
            env.enableCheckpointing(120 * 1000L, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(240 * 1000L);
            env.setParallelism(6);
            execute(env);
            env.execute(DASK_NAME);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void execute(StreamExecutionEnvironment env) throws Exception {
        final FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(
                "consumer-dws-signal-change",
//                ConfigLoader.get("kafka.clean.topic.dwd.ev", "dwd_clean_ev")
                ConfigLoader.get("kafka.topic.charge.detail", "dwm_charge_detail")

        );

//        kafkaConsumer.setStartFromTimestamp(DateUtil.parse("2023-06-01 20:00:55").getTime());
        DataStream<JSONObject> sourceData = env.addSource(kafkaConsumer).map(JSONUtil::parseObj);
        DataStream<JSONObject> result = sourceData
                .keyBy(k -> k.get(BaseFields.vin))
                .process(new SendOnChangeProcessFunction())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getLong(BaseFields.tid)))
                .keyBy(item -> item.getStr(BaseFields.vin))
                .window(TumblingEventTimeWindows.of(Time.minutes(2)))
                .apply(new ChargeProcessWindowFunction())
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    private transient JedisCluster jedisCluster;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
//                        jedisCluster = RedisClient.jedisCluster;
                    }

                    @Override
                    public void processElement(JSONObject nowData, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        log.info("set redis for vin: " + nowData.getStr("deviceId"));
                        RedisClient.setData(DWS_CHARGE_STATUS_KEY + nowData.getStr("deviceId"), nowData.toStringPretty(), DWS_CHARGE_STATUS_EXPIRE_KEY);
                    }
                });
        result.print();
    }

    private static class ChargeProcessWindowFunction implements WindowFunction<JSONObject, JSONObject, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<JSONObject> input, Collector<JSONObject> out) throws Exception {
            List<JSONObject> list = new ArrayList<>();
            Iterator<JSONObject> iterator = input.iterator();
            while (iterator.hasNext()) {
                JSONObject next = iterator.next();
                list.add(next);
            }
            list.sort(new Comparator<JSONObject>() {
                @Override
                public int compare(JSONObject o1, JSONObject o2) {
                    return o1.getStr(BaseFields.tid).compareTo(o2.getStr(BaseFields.tid));
                }
            });
            log.info("apply list size :{} list: {}", list.size(), list);
            out.collect(list.get(list.size() - 1));
        }
    }

    private static class SendOnChangeProcessFunction extends ProcessFunction<JSONObject, JSONObject> {
        private transient ValueState<String> bmsSoc;
        private transient ValueState<Boolean> isEdit;


        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 初始化 State
            bmsSoc = getRuntimeContext().getState(new ValueStateDescriptor<>("BMS_SOC", Types.STRING));
            isEdit = getRuntimeContext().getState(new ValueStateDescriptor<>("edit", Types.BOOLEAN));
        }

        @Override
        public void processElement(JSONObject nowData, Context ctx, Collector<JSONObject> out) throws Exception {
            String nowBms = nowData.getStr("BMS_SOC");
            if (ObjectUtil.isEmpty(bmsSoc.value())) {
                bmsSoc.update(nowBms);
            }
            isEdit.update(Boolean.FALSE);
            // 获取上一次的值
            String preBms = bmsSoc.value();
            if (preBms != null && !BaseFields.defaultVal.equals(nowBms) && !nowBms.equals(bmsSoc.value())) {
                bmsSoc.update(nowBms);
                isEdit.update(Boolean.TRUE);
            }

            if (isEdit.value()) {
                log.info("apply value :{} item: {}", nowData.getStr("deviceId"), nowData.getStr("item_time"));
//                System.out.printf("isEdit" + isEdit.value() + " vin: " + nowData.getStr("deviceId") + " bms: " + nowBms + " \n ");
                out.collect(createChargeData(nowData));
            }
        }

        private JSONObject createChargeData(JSONObject nowData) {
            JSONObject nowMap = new JSONObject();
            nowMap.put("deviceId", nowData.getStr("deviceId"));
            nowMap.put("BMS_SOC", nowData.getBigDecimal("BMS_SOC"));
            nowMap.put("BMS_ChrgSts", nowData.getInt("BMS_ChrgSts"));
            nowMap.put("BMS_BattCurr_RM", nowData.getBigDecimal("BMS_BattCurr_RM"));
            nowMap.put("SOH_COR", nowData.getInt("SOH_COR"));
            nowMap.put("BMS_ChrgTime", nowData.getLong("BMS_ChrgTime"));
            nowMap.put("item_time", nowData.getLong("item_time"));
            return nowMap;
        }
    }


}
